package com.huonan.component.thread;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.io.Serializable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * 异步线程 CompletableFuture Api使用
 */
@Slf4j
public class CompletableFutureOperations {

    /**
     * 异步计算：
     * runAsync ：没有返回结果
     * supplyAsync: 有返回结果
     */
    @Test
    public void asyncNoRt() {
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                //第一次请求：模拟耗时请求
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "result2";
        });
        log.info(future1.join() + ":" + future2.join());
    }

    /**
     * 1.阻塞当前线程直到计算完成；
     * 2.***如果调用者线程中断join正常返回get会抛出异常***
     * join() ：有返回结果，接收异常自己处理;
     * get(): 有返回结果,抛出异常
     */
    @Test
    public void asyncGetAndJoin() {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            // 模拟一个耗时的异步计算
            try {
                Thread.sleep(5000); // 等待2秒
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "result";
        });
        //主线程中断模拟
        Thread.currentThread().interrupt();
        try {
            // 使用get()方法等待异步计算完成并获取结果
            String result = future.get();
//            String result = future.join();
            log.info("Get()方法获取到结果：" + result);
            System.exit(0);
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage());
        }
    }

    /**
     * 异步计算：有返回结果
     * 1.thenAccept:只Accept（接收/返回）异步结果，不对异步结果进行操作
     * thenApply：对异步结果进行Apply（使用/处理）
     * 2.thenAccept和thenAcceptAsync:
     * 都对异步结果进行处理，不同是处理方式一个是同步一个是异步
     * thenAccept方法立即传入Consumer函数执行，thenAcceptAsync 是异步执行
     * 3. exceptionally
     * 接收异常，并返回
     */
    @Test
    public void asyncRt() throws InterruptedException {
        CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(() -> {
            // 模拟一个耗时的异步计算
            try {
                Thread.sleep(5000); // 等待2秒
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "result";
        });
        // 第二次请求：在异步计算完成之前，可以执行其他任务
        log.info("异步计算开始...");
        completableFuture.thenAccept(result -> {
            //对异步结果进行处理，没有返回
            log.info("异步返回结果：" + result);
        });
        CompletableFuture<String> thenApplyAsync = completableFuture.thenApply(result -> {
            //对异步结果进行处理，并返回
            return result.toUpperCase();
        });
        CompletableFuture<String> exceptionally = completableFuture.exceptionally(ex -> {
            //接收异常，并有返回
            log.info("异常返回：" + ex.getMessage());
            return "Error occurred ,but we recovered";
        });
    }

    /**
     * exceptionally: 异步计算返回异常，对异常进行处理
     */
    @Test
    public void asyncCalculateEx() throws ExecutionException, InterruptedException {
        CompletableFuture<Object> completableFuture = CompletableFuture.supplyAsync(() -> {
            //模拟异常
            throw new RuntimeException("模拟异常");
        });
        CompletableFuture<Object> exceptionally = completableFuture.exceptionally(ex -> {
            log.info("异常返回：" + ex.getMessage());
            return "Error occurred ,but we recovered";
        });
        // 异步计算完成前，模拟执行其它操作
        log.info("异步计算开始。。。");
        log.info("Result:" + exceptionally.get());
    }

    /**
     * allOf: 等待多个异步任务结果全部返回在执行下一步
     * anyOf: 多个任务只要有一个返回结果或者异常就结束
     */
    @Test
    public void asyncAllOf() {
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
            // CompletableFuture 实例1
            try {
                Thread.sleep(4000);
                log.info("Running task 1...");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
            // CompletableFuture 实例2
            try {
                Thread.sleep(2000);
                log.info("Running task 2...");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        //allOf 全部返回
        CompletableFuture<Void> allOf = CompletableFuture.allOf(future1, future2);
        log.info("异步子任务执行开始");
        allOf.join();//等待所有结果返回，如果有任一个抛出异常结束
        log.info("全部执行完成");
        //返回最新执行完的
        CompletableFuture<Object> anyOf = CompletableFuture.anyOf(future1, future2);
        log.info("异步子任务执行开始");
        anyOf.thenAccept(result -> {
            log.info("At least one future completed.");
        }).join();
    }


    /**
     * thenCombine 两个异步任务结果进行组合
     */
    @Test
    public void ansCombine() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "result1";
        });

        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> {
            return 40;
        });
        CompletableFuture<Object> completableFuture1 = future1.thenCombine(future2, (r1, r2) -> {
            //结果进行组合
            return (r1 + r2);
        });
        log.info("组合结果：" + completableFuture1.get());
    }


    /**
     * whenComplete，extractHandle : 异步计算回调
     * whenComplete:不需要返回结果
     * extractHandle：需要返回结果
     */
    @Test
    public void ansWhenComplete() throws InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "result1";
        });
        /*whenComplete*/
//        CompletableFuture<String> complete = future1.whenComplete((result, t) -> {
//            if (t == null) {
//                log.info("Future completed normally with result: " + result.toUpperCase());
//            } else {
//                log.info("Future completed exceptionally with cause: " + t.getCause());
//            }
//        });
//        System.out.println("callback：" + complete.join());

        /*extractHandle*/
        CompletableFuture<Serializable> handle = future1.handle((result, t) -> {
            if (t == null) {
                return result.toUpperCase();
            } else {
                log.info("Future completed exceptionally with cause: " + t.getCause());
                return t.getCause();
            }
        });
        System.out.println("callback：" + handle.join());
    }

    /**
     * acceptEither：用于对最先返回的结果进行处理,不再返回
     */
    @Test
    public void ansAcceptEither() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "result1";
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "result2";
        });

        /*acceptEither*/
        CompletableFuture<Void> future = future1.acceptEither(future2, result -> {
            //模拟业务处理
            log.info("First future completed with result: " + result.toUpperCase());
        });
        future.join();

    }

    /**
     * * thenCompose : 用于将一个计算任务的结果传递给下一个计算任务
     */
    @Test
    public void asyncThenCompose() {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return "result1";
        });
        CompletableFuture<String> result = future1.thenCompose(r1 -> {
            return CompletableFuture.supplyAsync(() -> {
                return r1.toUpperCase();
            });
        });
        log.info("计算结果："+result.join());
    }

    /**
     * runAfterBoth : 等待两个结果都返回
     * runAfterEither: 两个结果返回最先完成的
     */
    @Test
    public void asyncRunAfterBoth() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
                log.info("result1");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            return "result1";
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {

            try {
                Thread.sleep(1000);
                log.info("result2");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            return "result2";
        });
        CompletableFuture<Void> future3 = future1.runAfterBoth(future2, () -> {
            // 在两个异步任务都完成后执行的操作
            log.info("Both tasks completed");
        });
//        CompletableFuture<Void> future3 = future1.runAfterEither(future2, () -> {
//            // 在其中一个异步任务完成后执行的操作
//            log.info("One of the tasks completed");
//        });
        future3.join();
    }
}
